package dmp.beans.sparksql

import java.util.Properties

import com.typesafe.config.ConfigFactory
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * author:CN.CDG
  * Date:2019/2/15
  * Time:12:42
  **/
object MediaCt_Mysql {
  def main(args: Array[String]): Unit = {
    if(args.length!=2){
      println("参数设置不合法，退出程序")
      sys.exit()
    }
    val Array(inputPath,midinputPath)=args
    val conf=new SparkConf().setMaster("local[*]").setAppName("AreaAnalyseRpt")
      .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")//rdd序列化到磁盘，用于worker和worker之间的传输
    //"spark.serializer","org.apache.spark.serializer.KryoSerializer
    val spark=SparkSession.builder().config(conf).getOrCreate()
    //val sQLContext =new SQLContext(spark)

    //读取parquet文件
    val parquetData: DataFrame = spark.read.parquet(inputPath)
    val dsData = spark.read.textFile(midinputPath)
    dsData.createOrReplaceTempView("log2")
    //拿到数据后注册一张临时表
    parquetData.createOrReplaceTempView("log")
    val result: DataFrame = spark.sql(
     /* """
        | select
        |if (m.appname <> 'null' , m.appname, l.appname) as appname,
        |sum(case when requestmode = 1 and processnode >= 1 then 1 else 0 end) original_request_number,
        |sum(case when requestmode = 1 and processnode >= 2 then 1 else 0 end) valid_request_number,
        |sum(case when requestmode = 1 and processnode = 3 then 1 else 0 end) as_request_number,
        |sum(case when iseffective = 1 and isbilling = 1 and isbid = 1 then 1 else 0 end) bidding_count,
        |sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 and adorderid <> 0 then 1 else 0 end) bidding_v,
        |if (sum(case when iseffective = 1 and isbilling = 1 and isbid = 1 then 1 else 0 end) = 0 ,0.0,
        |sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 and adorderid <> 0 then 1 else 0 end)/sum(case when iseffective = 1 and isbilling = 1 and isbid = 1 then 1 else 0 end) ) as bidding_success,
        |sum(case when requestmode = 2 and iseffective = 1 then 1 else 0 end) ad_look_count,
        |sum(case when requestmode = 3 and iseffective = 1 then 1 else 0 end) ad_click_count,
        |if (sum(case when requestmode = 2 and iseffective = 1 then 1 else 0 end) = 0,0.0,sum(case when requestmode = 3 and iseffective = 1 then 1 else 0 end) / sum(case when requestmode = 2 and iseffective = 1 then 1 else 0 end)) as click_rate,
        |sum(case when adplatformproviderid >= 100000 and adcreativeid >= 200000 and adorderid >= 200000 and iseffective = 1 and isbilling = 1 and iswin = 1 and winprice > 0 then winprice/1000 else 0 end)/1000 as ad_cost,
        |sum(case when adplatformproviderid >= 100000 and adcreativeid >= 200000 and adorderid >= 200000 and iseffective = 1 and isbilling = 1 and iswin = 1 and adpayment > 0 then adpayment/1000 else 0 end) as ad_consume
        |from
        |log2 l
        |left join log m
        |on l.appid = m.appid
        |group by l.appname,m.appname
      """.stripMargin*/
      """
        |select appname,
        |sum(case when requestmode = 1 and processnode >=1 then 1 else 0 end) as `原始请求`,
        |sum(case when requestmode = 1 and processnode >=2 then 1 else 0 end) as `有效请求`,
        |sum(case when requestmode = 1 and processnode = 3 then 1 else 0 end) as `广告请求`,
        |sum(case when iseffective = 1 and isbilling = 1 and isbid = 1 then 1 else 0 end) as `参与竞价数`,
        |sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 and adorderid <> 0 then 1 else 0 end) as `竞价成功数`,
        |sum(case when iseffective = 1 and isbilling = 1 and isbid = 1 then 1 else 0 end)/sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 and adorderid <> 0 then 1 else 0 end) as `竞价成功率`,
        |sum(case when requestmode = 2 and iseffective = 1 then 1 else 0 end) as `展示量`,
        |sum(case when requestmode = 3 and iseffective = 1 then 1 else 0 end) as `点击量`,
        |sum(case when requestmode = 2 and iseffective = 1 then 1 else 0 end)/sum(case when requestmode = 3 and iseffective = 1 then 1 else 0 end) as `点击率`,
        |sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 then winprice else 0 end)/1000 as `广告成本`,
        |sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 then adpayment else 0 end)/1000 as `广告消费`
        |from log
        |group by appname
      """.stripMargin
    )
    val load=ConfigFactory.load()
    val props=new Properties()
    props.setProperty("user",load.getString("jdbc.user"))
    props.setProperty("password",load.getString("jdbc.password"))
    result.write.jdbc(load.getString("jdbc.url"),load.getString("jdbc.tablename7"),props)
    spark.stop()
  }
}
